/*
 * 21/10/2011 - 22:33
 */
package org.nettyflux.tcp;

import org.nettyflux.netmessages.NetworkMessage;
import org.nettyflux.mailbox.MailBox;
import java.io.BufferedInputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.net.InetAddress;
import java.net.Socket;
import org.nettyflux.communicator.ListeningThread;
import org.nettyflux.communicator.Receiver;
import org.nettyflux.debug.GlobalDebugger;
import org.nettyflux.mailbox.ReceiveListener;
import org.nettyflux.netmessages.MessageReceived;
import org.nettyflux.netmessages.SendCommand;

/**
 * @author Victor Williams Stafusa da Silva
 */
public class TCPClientCommunicator implements Receiver<MessageReceived<TCPHost>> {

    private final Socket socket;
    private final OutputStream output;
    private final BufferedInputStream input;
    private final TCPHost client;
    private final MailBox<NetworkMessage> mailbox;
    private final TCPClientSenderListener senderListener;
    private final TCPClientCloseListener closeListener;
    private ListeningThread listeningThread;

    public TCPClientCommunicator(MailBox<NetworkMessage> mailbox, String address, int port, boolean autoReceive) throws IOException {
        this(mailbox, InetAddress.getByName(address), port, autoReceive);
    }

    public TCPClientCommunicator(MailBox<NetworkMessage> mailbox, InetAddress address, int port, boolean autoReceive) throws IOException {
        this(mailbox, new Socket(address, port), autoReceive);
    }

    public TCPClientCommunicator(MailBox<NetworkMessage> mailbox, TCPHost client, boolean autoReceive) throws IOException {
        this(mailbox, new Socket(client.getAddress(), client.getPort()), client, autoReceive);
    }

    public TCPClientCommunicator(MailBox<NetworkMessage> mailbox, Socket socket, boolean autoReceive) throws IOException {
        this(mailbox, socket, new TCPHost(socket.getInetAddress(), socket.getPort()), autoReceive);
    }

    private TCPClientCommunicator(MailBox<NetworkMessage> mailbox, Socket socket, TCPHost client, boolean autoReceive) throws IOException {
        this.socket = socket;
        this.output = socket.getOutputStream();
        this.input = new BufferedInputStream(socket.getInputStream());
        this.client = client;
        this.mailbox = mailbox;
        this.senderListener = new TCPClientSenderListener();
        this.closeListener = new TCPClientCloseListener();
        this.mailbox.addReceiveListener(SendCommand.class, senderListener);
        this.mailbox.addReceiveListener(TCPCloseConnectionCommand.class, closeListener);
        if (autoReceive) ListeningThread.create(this);
    }

    public synchronized void setEagerMode(boolean eagerMode) {
        if ((listeningThread != null) == eagerMode) return;
        if (eagerMode) {
            listeningThread = ListeningThread.create(this);
        }
    }

    public Socket getSocket() {
        return socket;
    }

    public void send(byte[] message) throws IOException {
        if (isClosed()) throw new IllegalStateException("The socket is closed.");

        if (GlobalDebugger.isEnabled()) {
            GlobalDebugger.println("Sending TCP: " + java.util.Arrays.toString(message));
        }

        output.write(message);
        output.flush();
    }

    @Override
    public MessageReceived<TCPHost> receive() throws IOException {
        if (isClosed()) throw new IllegalStateException("The socket is closed.");

        if (GlobalDebugger.isEnabled()) {
            GlobalDebugger.println("Will receive TCP.");
        }

        SimpleByteBuffer msg = new SimpleByteBuffer();
        msg.appendAvailable(input);

        if (GlobalDebugger.isEnabled()) {
            GlobalDebugger.println("Received TCP: " + java.util.Arrays.toString(msg.getBytes()));
        }

        MessageReceived<TCPHost> message = MessageReceived.create(client, msg.getBytes());
        mailbox.postMessage(message);
        return message;
    }

    @Override
    public void close() throws IOException {
        mailbox.removeReceiveListener(senderListener);
        mailbox.removeReceiveListener(closeListener);
        socket.close();
    }

    @Override
    public boolean isClosed() {
        return socket.isClosed();
    }

    public TCPHost getClient() {
        return client;
    }

    private class TCPClientCloseListener implements ReceiveListener<TCPCloseConnectionCommand> {

        @Override
        public void receive(TCPCloseConnectionCommand message) {
            if (message.getPort() == client.getPort()) {
                try {
                    close();
                    mailbox.postMessage(message.makeAck());
                } catch (IOException e) {
                    mailbox.postMessage(message.makeFailAck(e));
                }
            }
        }
    }

    private class TCPClientSenderListener implements ReceiveListener<SendCommand<?>> {

        @Override
        public void receive(SendCommand<?> message) {
            if (!(message.getReceiver() instanceof TCPHost)) return;
            TCPHost host = (TCPHost) message.getReceiver();
            if (host.equals(client)) {
                try {
                    send(message.getMessage());
                    mailbox.postMessage(message.makeAck());
                } catch (IOException e) {
                    mailbox.postMessage(message.makeFailAck(e));
                }
            }
        }
    }
}
